import collections
import datetime
import json
import threading
import time
import unicodedata

import tushare as ts
import pandas as pd
import numpy as np
from django.db.models import Count, F
from django.utils.functional import classproperty

from config import *
from processor.constants import *
from processor.models import *
from processor.utils import get_last_month, get_pre_quarter, tushare_batch_fetch

if USING_WIND:
    if OS == Linux:
        from WindLinuxAPI.WindPy import w
    else:
        from WindPy import w

ts.set_token(TUSHARE_TOKEN)
TS_PRO = ts.pro_api(timeout=300)

class TushareAPI:
    keep_running = False
    update_queue = []

    _lazy_load_stock_codes = {
        'data': StockBasic.objects.values_list('scode', flat=True),
        'datetime': datetime.datetime.now(),
    }

    @classproperty
    def STOCK_CODES(cls):
        if cls._lazy_load_stock_codes is not None:
            if (datetime.datetime.now() - cls._lazy_load_stock_codes['datetime']).days == 0:
                return cls._lazy_load_stock_codes['data']
        
        cls._lazy_load_stock_codes = {
            'data': TS_PRO.stock_basic(fields='ts_code').ts_code.to_list(),
            'datetime': datetime.datetime.now(),
        }

        return cls._lazy_load_stock_codes['data']

    @classmethod
    def update_stock_list(cls, scode="ALL", callback=print):
        '''
            方法： 更新股票列表
                参数：
                    scode: 股票代码
                        1. 传入[scode]: 更新该股票的基础信息
                        2. 留空则更新股票列表中所有股票
        '''

        if not scode or scode == "ALL":
            created_count, stock_count = 0, len(cls.STOCK_CODES)
            if callback is not None: callback(0, stock_count)
            for code in cls.STOCK_CODES:
                try:
                    for _ in range(RETRY_COUNT):
                        created = StockBasic.update_by_code(code)
                        created_count += created
                        if callback is not None: callback(created_count, stock_count)
                        break
                except Exception as e:
                    Logger.debug(msg=f"Stock {code} update failed: {e}", module='crawler')
                    time.sleep(RETRY_TIME_DELAY)
            Logger.info(msg=f"列表更新完成，本次新增 {created_count} 只股票记录。", module='crawler')
        else:
            if callback is not None: callback(0, 1)

            created = StockBasic.update_by_code(scode)
            if callback is not None: callback(1, 1)
            Logger.info(msg=f"列表更新完成，本次新增 {int(created)} 只股票记录。", module='crawler')

    @classmethod
    def update_all_stock_basic(cls, start_date=None, end_date=None, callback = print):
        '''
            方法:  更新列表中所有股票数据
                全局变量 (引用自 config 文件):
                    THREADS: 线程数量
                参数:
                    start_date: 开始日期
                    end_date: 结束日期
                    callback: 进度更新回调函数
        '''
        assert (start_date and end_date) or (not start_date and not end_date), "Illegal start_date or end_date."
        stock_count = len(cls.STOCK_CODES)
        if callback is not None: callback(0, stock_count)
        cls.keep_running = True
        cls.update_queue = cls.STOCK_CODES[:]
        for _ in range(THREADS):
            threading.Thread(target=cls.update_stock_basic, args=(start_date, end_date, )).start()
        if start_date and end_date:
            Logger.info(msg=f"Tushare 源更新已启动，本次更新起止日期：{start_date}~{end_date}。", module='crawler')
        else:
            Logger.info(msg=f"Tushare 源更新已启动，本次更新至：{datetime.date.today()}。", module='crawler')
        while cls.keep_running and cls.update_queue:
            if callback is not None: callback(stock_count - len(cls.update_queue), stock_count)
            time.sleep(1)
        cls.keep_running = False
        cls.update_queue.clear()
        Logger.info(msg=f"Tushare 源更新完成，本次更新 {stock_count - len(cls.update_queue)} 只股票记录。", module='crawler')

    @classmethod
    def update_stock_basic(cls, start_date=None, end_date=None):
        '''
            子进程方法:  更新列表中所有股票数据(由 update_all_stock_basic 调用)
        '''
        while cls.keep_running and cls.update_queue:
            code = cls.update_queue.pop()
            for period in ('weekly', 'daily'):
                for _ in range(RETRY_COUNT):
                    try:
                        #拉取该股票的基础信息记录
                        stock_basic = StockBasic.get_by_code(code)
                        if stock_basic is None: break

                        if start_date and end_date:
                            # 指定日期拉取
                            last_update = datetime.datetime.strptime(str(start_date), "%Y-%m-%d").date()
                            # 拉取 Tushare 前复权的 收盘价、MA5、MA10、MA20 (为计算MA20需设定日期的提前量)
                            stock_data = ts.pro_bar(
                                code, freq=period[0].upper(), adj="qfq", ma=[5,10,20], start_date = str(last_update - datetime.timedelta(days = 40)), end_date = str(end_date)
                            )
                        else:
                            # 不指定日期时, 更新至最新
                            last_update = (datetime.datetime.strptime(max(stock_basic.tushare_index + [DEFAULT_START_DATE]), "%Y-%m-%d") + datetime.timedelta(days=1)).date()
                            
                            # 已更新到最新时跳过
                            if last_update >= datetime.date.today(): break
                            # 拉取 Tushare 前复权的 收盘价、MA5、MA10、MA20 (为计算MA20需设定日期的提前量)
                            stock_data = ts.pro_bar(
                                code, freq=period[0].upper(), adj="qfq", ma=[5,10,20], start_date = str(last_update - datetime.timedelta(days = 40))
                            )

                        if stock_data is None or stock_data.empty: break
                        
                        # 取交易日期、收盘价、均线 
                        # 去除未满20天的数据
                        stock_data = stock_data.loc[:, 
                            ['trade_date', 'close', 'ma5', 'ma10', 'ma20']
                        ].dropna()
                        
                        if stock_data.empty: break

                        # 格式化日期为 2020-12-01 形式
                        stock_data['trade_date'] = stock_data.apply(
                            lambda x: x['trade_date'][:4] + '-' + x['trade_date'][4:6] + '-' + x['trade_date'][6:], axis=1
                        )
                        # 转为 Dict 进行存储
                        stock_data = stock_data.set_index('trade_date').to_dict(orient = 'index')
                        # 去除停牌日
                        for k in stock_basic.suspension:
                            if k in stock_data:
                                stock_data.pop(k)
                        # 去除提前量
                        for k in list(stock_data.keys()):
                            if k <= str(last_update):
                                stock_data.pop(k)
                        # 过滤今日收盘前数据
                        if start_date and end_date:
                            if stock_basic.last_update < last_update:
                                stock_basic.last_update = last_update
                        elif datetime.datetime.now().hour < 15:
                            if str(datetime.date.today()) in stock_data:
                                stock_data.pop(str(datetime.date.today()))
                            stock_basic.last_update = datetime.date.today() - datetime.timedelta(days = 1)
                        else:
                            stock_basic.last_update = datetime.date.today()
                        # 筛选后数据为空时跳出
                        if not stock_data: 
                            stock_basic.save(update_fields=['last_update', ])
                            break

                        # 写入日级数据
                        if period == 'daily':
                            # 增加索引至记录，同时进行排序
                            stock_basic.tushare_index = sorted(set(list(stock_data.keys()) + stock_basic.tushare_index))
                            # 更新数据
                            stock_basic.tushare_source.update(stock_data)
                            stock_basic.save(update_fields=['last_update', 'tushare_index', 'tushare_source'])
                        # 写入周级数据
                        else:
                            # 增加索引至记录，同时进行排序
                            stock_basic.tushare_weekly_index = sorted(set(list(stock_data.keys()) + stock_basic.tushare_weekly_index))
                            # 更新数据
                            stock_basic.tushare_weekly_source.update(stock_data)
                            stock_basic.save(update_fields=['last_update', 'tushare_weekly_index', 'tushare_weekly_source'])

                        break
                    except Exception as e:
                        Logger.debug(msg=f"Stock {code} update failed, Source: Tushare, {e}", module='crawler')
                        time.sleep(RETRY_TIME_DELAY)

    @classmethod
    def update_tushare_rsi(cls, start_date=None, end_date=None, callback=print):
        '''
            方法:  更新Tushare官方RSI数据
                参数:
                    start_date: 开始日期
                    end_date: 结束日期
                    callback: 进度更新回调函数
        '''
        assert (start_date and end_date), "Illegal start_date or end_date."
        start_date = date_format(start_date, datetime.date)
        end_date = date_format(end_date, datetime.date)
        Logger.info(msg=f"Tushare 官方RSI数据更新已启动，本次更新起止日期：{start_date}~{end_date}。", module='crawler')
        
        # 天数及当前增量
        days = (end_date - start_date).days
        offset = 0

        df = None
        while offset <= days:
            # 进度回调
            callback(offset, days)
            # 拉取指定日期数据
            data = TS_PRO.stk_factor_pro(
                trade_date=(start_date + datetime.timedelta(days=offset)).strftime('%Y%m%d'),
                fields='trade_date,ts_code,rsi_qfq_6', 
            )
            offset += 1
            if data.empty: continue
            elif df is None: df = data
            else: df = pd.concat([df, data])

        # 无数据
        if df is None: return

        # 格式化日期
        df['trade_date'] = df['trade_date'].apply(
            lambda x: x[:4] + '-' + x[4:6] + '-' + x[6:]
        )

        # 去除 NaN
        df = df.dropna()
        
        stock_count = len(set(df.ts_code))
        for idx, scode in enumerate(set(df.ts_code)):
            # 进度回调
            callback(idx, stock_count)

            # 获取 StockRSI
            stock_rsi = StockRSI.objects.filter(scode=scode).first()
            if stock_rsi is None: continue

            # 转为字典存储
            rsi_data = df[df.ts_code == scode].set_index('trade_date')['rsi_qfq_6'].to_dict()

            # 写入数据库
            stock_rsi.rsi.update(rsi_data)
            stock_rsi.rsi_index = sorted(stock_rsi.rsi.keys())
            stock_rsi.save()

        Logger.info(msg=f"Tushare 官方RSI数据更新完成，本次更新 {stock_count} 只股票记录。", module='crawler')
        

    @classmethod
    def update_all_stock_rsi(cls, period='daily', rsi_n=6, callback=print):
        '''
            方法:  计算并覆盖列表中所有股票RSI数据
                全局变量 (引用自 config 文件):
                    THREADS: 线程数量
                参数:
                    callback: 进度更新回调函数
        '''
        stock_count = len(cls.STOCK_CODES)
        callback and callback(0, stock_count * 2)

        Logger.info(msg="Tushare RSI 计算已启动。", module='crawler')

        cls.keep_running = True
        cls.update_queue = cls.STOCK_CODES[:]

        for _ in range(THREADS):
            threading.Thread(target=cls.update_stock_rsi, args=(rsi_n, period), ).start()
        
        while cls.keep_running and cls.update_queue:
            callback and callback(stock_count - len(cls.update_queue), stock_count * 2)
            time.sleep(1)

        cls.keep_running = False
        cls.update_queue.clear()

        Logger.info(msg=f"Tushare RSI 更新完成， 本次更新{stock_count}条股票记录。", module='crawler')


    @classmethod
    def update_weekly_all_stock_rsi(cls, rsi_n=6, callback=print):
        '''
            方法:  计算并覆盖列表中所有周级股票RSI数据
                全局变量 (引用自 config 文件):
                    THREADS: 线程数量
                参数:
                    callback: 进度更新回调函数
        '''
        return cls.update_all_stock_rsi('weekly', rsi_n, callback)


    @classmethod
    def update_stock_rsi(cls, rsi_n=6, period='daily'):
        '''
            子进程方法： 更新列表中所有股票RSI数据
                参数：
                    period: 周期
        '''
        while cls.keep_running and cls.update_queue:
            code = cls.update_queue.pop()
            try:
                # 拉取该股票的RSI记录
                stock_rsi = StockRSI.get_by_code(code)
                if stock_rsi is None: break
                stock_basic = stock_rsi.stock_basic

                # 将基础数据载入DF
                if period == 'daily':
                    df = pd.DataFrame(stock_basic.tushare_source, index=['close']).T
                else:
                    df = pd.DataFrame(stock_basic.tushare_weekly_source, index=['close']).T

                df = df.sort_index()
                df['preclose'] = df['close'].shift(1)

                # 计算每日差值
                df['diff'] = (df['close'] - df['preclose']).abs()

                # 计算每日增值
                df['raise'] = df['close'] - df['preclose']
                df.loc[df['raise'] < 0, 'raise'] = 0

                # 计算差值与增值的 SMA （加权移动平均）
                df['diff_sma'] = df['diff'].ewm(com=rsi_n-1, adjust=False).mean()
                df['raise_sma'] = df['raise'].ewm(com=rsi_n-1, adjust=False).mean()

                # 计算 RSI
                df['rsi'] = 100 * df['raise_sma'] / df['diff_sma']

                # 保存至数据库
                rsi_data = df['rsi'].dropna().to_dict()

                if period == 'daily':
                    stock_rsi.rsi = rsi_data
                    stock_rsi.rsi_index = sorted(rsi_data.keys())
                else:
                    stock_rsi.wrsi = rsi_data
                    stock_rsi.wrsi_index = sorted(rsi_data.keys())
                stock_rsi.save()
    
            except Exception as e:
                Logger.debug(msg=f"Stock {code} RSI update failed, {e}", module='crawler')

    
    @classmethod
    def update_all_stock_top10_holder(cls, callback=print):
        '''
            方法: 计算并覆盖列表中所有股票TOP10股东数据
                全局变量 (引用自 config 文件):
                    THREADS: 线程数量
                参数: 
                    callback: 进度更新回调函数
        '''
        stock_count = len(cls.STOCK_CODES)
        callback and callback(0, stock_count * 2)

        Logger.info(msg="Tushare TOP10股东 计算已启动。", module='crawler')

        cls.keep_running = True
        cls.update_queue = cls.STOCK_CODES[:]

        Shareholding.objects.truncate()

        for _ in range(THREADS):
            threading.Thread(target=cls.update_stock_top10_holder, args=(), ).start()
        
        while cls.keep_running and cls.update_queue:
            callback and callback(stock_count - len(cls.update_queue), stock_count)
            time.sleep(1)
        
        threading.Thread(target=cls.update_top_1000_shareholding, args=(), ).start()

        cls.keep_running = False
        cls.update_queue.clear()
        
        Logger.info(msg=f"Tushare TOP10股东 更新完成， 本次更新{stock_count}条股票记录。", module='crawler')


    @classmethod
    def update_stock_top10_holder(cls, ):
        while cls.keep_running and cls.update_queue:
            code = cls.update_queue.pop()
            try:
                # year = datetime.datetime.now().strftime('%Y')
                # 获取 StockBasic 对象
                stock = StockBasic.get_by_code(code)

                # 获取股票所属概念
                concepts = '/'.join(stock.concepts.values_list('cname', flat=True))

                for retry in range(RETRY_COUNT):
                    try:
                        # QPS: 400 / 60
                        df = TS_PRO.top10_floatholders(ts_code=code)
                        break
                    except Exception as e:
                        Logger.debug(msg=f"Stock {code} Top 10 Holder fetch failed, retrying: {e}", module='crawler')

                        time.sleep(RETRY_TIME_DELAY)
                        if retry + 1 == RETRY_COUNT:
                            raise e
                        
                # 圆角转半角
                df.holder_name = df.holder_name.map(lambda x: unicodedata.normalize('NFKC', x))

                # 去除脏数据
                df = df[
                    df.end_date.str.endswith('0331') | 
                    df.end_date.str.endswith('0630') | 
                    df.end_date.str.endswith('0930') | 
                    df.end_date.str.endswith('1231')
                ]

                # 排序
                df = df.sort_values(by=['end_date', 'hold_amount'], ascending=[False, False])
                # 去重
                df = df.drop_duplicates(subset=['end_date', 'holder_name'])
                # 将日期转为季度
                df['quarter'] = pd.PeriodIndex(df.end_date, freq='Q')

                # 处理流通占比为 NaN
                df = df[~df.hold_float_ratio.isna()]
                # df.loc[df.hold_float_ratio.isna(), 'hold_float_ratio'] = -100

                # 变更类型 减少/不变/增加
                df['change_type'] = np.sign(df.hold_change.fillna(np.nan))
                
                # 新增部分 修复变动数量
                df.loc[df.hold_change.isna(), 'change_type'] = 2
                df.loc[df.hold_change.isna(), 'hold_change'] = df.hold_amount

                # 变更比例
                df['hold_float_ratio_change'] = df.hold_change * df.hold_float_ratio / df.hold_amount
                df['hold_float_ratio_change'] = df['hold_float_ratio_change'].fillna(0)

                # 批量保存
                Shareholding.objects.bulk_create([
                    Shareholding(
                        scode=code,
                        sname=stock.sname,
                        industry=stock.industry,
                        stock_basic=stock,
                        concepts=concepts,
                        quarter=hold_info.quarter,
                        is_personal=len(hold_info.holder_name) <= 4,
                        holder=hold_info.holder_name,
                        change_type=hold_info.change_type,
                        prop_change=hold_info.hold_float_ratio_change,
                        hold_prop=hold_info.hold_float_ratio,
                        hold_amount=hold_info.hold_amount,
                        prev_hold_prop=hold_info.hold_float_ratio - hold_info.hold_float_ratio_change,
                        prev_hold_amount=hold_info.hold_amount - hold_info.hold_change,
                    )
                    for _, hold_info in df.iterrows()
                ], batch_size=64)

            except Exception as e:
                Logger.debug(msg=f"Stock {code} TOP10 Holder update failed, {e}", module='crawler')


    @classmethod
    def update_top_1000_shareholding(cls, ):
        try:
            # 初始化
            Shareholding.objects.update(is_legend=False)
            # 更新季度 牛散持股人
            for quarter in Shareholding.objects.values_list('quarter', flat=True).distinct():
                # 持股数量最大的 前 1000 名
                top_1000_legend_holders = list(Shareholding.objects.filter(
                    quarter=quarter,
                    is_personal=True,
                ).values(
                    'holder'
                ).annotate(
                    count=Count(1)
                ).order_by(
                    '-count'
                ).values_list(
                    'holder', flat=True
                )[:1000])
                # 更新为 牛散
                Shareholding.objects.filter(
                    quarter=quarter,
                    holder__in=top_1000_legend_holders
                ).update(
                    is_legend=True
                )
        except Exception as e:
            Logger.error(msg=f"Stock TOP1000 Legend Holder update failed, {e}", module='crawler')


    @classmethod
    def update_all_stock_holder_number(cls, callback=print):
        '''
            方法: 计算并覆盖列表中所有股票股东户数数据
                全局变量 (引用自 config 文件):
                    THREADS: 线程数量
                参数: 
                    callback: 进度更新回调函数
        '''
        stock_count = len(cls.STOCK_CODES)
        callback and callback(0, stock_count * 2)

        Logger.info(msg="Tushare 股东户数 计算已启动。", module='crawler')

        cls.keep_running = True
        cls.update_queue = cls.STOCK_CODES[:]

        for _ in range(THREADS):
            threading.Thread(target=cls.update_stock_holder_number, args=(), ).start()
        
        while cls.keep_running and cls.update_queue:
            callback and callback(stock_count - len(cls.update_queue), stock_count)
            time.sleep(1)

        cls.keep_running = False
        cls.update_queue.clear()
        
        Logger.info(msg=f"Tushare 股东户数 更新完成， 本次更新 {stock_count} 条股票记录。", module='crawler')

    
    @classmethod
    def update_stock_holder_number(cls, ):
        while cls.keep_running and cls.update_queue:
            code = cls.update_queue.pop()
            try:
                stock = StockBasic.get_by_code(code)

                for _ in range(RETRY_COUNT):
                    try:
                        # QPS: 400 / 60
                        df = TS_PRO.stk_holdernumber(ts_code=code)
                        break
                    except:
                        time.sleep(RETRY_TIME_DELAY)
                
                if len(df) < 1: continue

                # 处理日期
                df = df.dropna().drop(['ts_code', 'ann_date',], axis=1)
                df['end_date'] = df.apply(
                    lambda x: x['end_date'][:4] + '-' + x['end_date'][4:6] + '-' + x['end_date'][6:], axis=1
                )

                # 筛选季度数据并去重
                pdf = df[
                    df.end_date.str.endswith('03-31') | 
                    df.end_date.str.endswith('06-30') | 
                    df.end_date.str.endswith('09-30') | 
                    df.end_date.str.endswith('12-31')
                ]
                pdf = pdf.drop_duplicates(subset=['end_date'])
                
                # 计算周期/季度数据的变化量/比例
                df['prop'] = (df.holder_num.iloc[0] / df.holder_num - 1) * 100
                pdf['prop'] = (df.holder_num.iloc[0] / pdf.holder_num - 1) * 100

                # 本周期数据
                current_period = df.iloc[0]
                # 历史季度数据
                pdf = pdf[pdf.end_date < current_period.end_date]
                
                p30d, p60d = date_offset(current_period.end_date, -30), date_offset(current_period.end_date, -60)
                df['p30d'] = df.end_date.map(lambda x: abs(date_diff(x, p30d)))
                df['p60d'] = df.end_date.map(lambda x: abs(date_diff(x, p60d)))

                p30df = df.sort_values('p30d')
                p60df = df.sort_values('p60d')

                safe_iloc = lambda series, index: series.iloc[index] if index < len(series) else None

                StockHolderNumber.objects.update_or_create(
                    stock_basic=stock,
                    defaults={
                        'scode': stock.scode, 
                        'sname': stock.sname,
                        'industry': stock.industry,
                        't_0_d': safe_iloc(df.end_date, 0),
                        't_0_c': safe_iloc(df.holder_num, 0),
                        't_q_d': safe_iloc(df.end_date, 1),
                        't_q_c': safe_iloc(df.holder_num, 1),
                        't_q_r': safe_iloc(df.prop, 1),
                        't_30_d': safe_iloc(p30df.end_date, 0),
                        't_30_c': safe_iloc(p30df.holder_num, 0),
                        't_30_r': safe_iloc(p30df.prop, 0),
                        't_60_d': safe_iloc(p60df.end_date, 0),
                        't_60_c': safe_iloc(p60df.holder_num, 0),
                        't_60_r': safe_iloc(p60df.prop, 0),
                        't_90_d': safe_iloc(pdf.end_date, 1),
                        't_90_c': safe_iloc(pdf.holder_num, 1),
                        't_90_r': safe_iloc(pdf.prop, 1),
                        't_180_d': safe_iloc(pdf.end_date, 2),
                        't_180_c': safe_iloc(pdf.holder_num, 2),
                        't_180_r': safe_iloc(pdf.prop, 2),
                        't_360_d': safe_iloc(pdf.end_date, 4),
                        't_360_c': safe_iloc(pdf.holder_num, 4),
                        't_360_r': safe_iloc(pdf.prop, 4),
                    }
                )

            except Exception as e:
                Logger.debug(msg=f"Stock {code} Holder Number update failed, {e}", module='crawler')


    @classmethod
    def update_all_stock_achievement(cls, period, callback=print):
        '''
            方法: 计算并覆盖指定周期的所有股票业绩数据
                参数: 
                    period: 周期
                    callback: 进度更新回调函数
        '''
        Logger.info(msg="Tushare 业绩筛选 更新已启动。", module='crawler')

        if isinstance(period, datetime.date):
            period = str(period)
        else:
            try:
                if period in get_quarter_periods():
                    period = period.replace('Q1', '0331').replace('Q2', '0630').replace('Q3', '0930').replace('Y', '1231')
                else:
                    datetime.datetime.strptime(period, '%Y%m%d')
            except Exception as e:
                return Logger.debug(msg=f"Illegal Period {period}, {e}", module='crawler')
        
        quarter = period.replace('0331', 'Q1').replace('0630', 'Q2').replace('0930', 'Q3').replace('1231', 'Y')

        # 周期不合法时退出
        if not any(period.endswith(tail) for tail in ('0331', '0630', '0930', '1231')):
            return Logger.warning(msg=f"Illegal Period {period}.", module='crawler')

        # 拉取利润表数据
        income = TS_PRO.income_vip(period=period, fields='ts_code,ann_date,revenue,total_cogs,update_flag', limit=2**16)
        # 去重 + 补0
        income = income.sort_values(by=['ts_code', 'update_flag'], ascending=[True, False])
        income = income.drop_duplicates(subset=['ts_code']).drop('update_flag', axis=1)

        # 拉取财务指标数据
        updated_fina = TS_PRO.fina_indicator_vip(period=period, update_flag='1', fields='ts_code,roe,profit_dedt,profit_to_gr,assets_turn,saleexp_to_gr,update_flag')
        fina = TS_PRO.fina_indicator_vip(period=period, update_flag='0', fields='ts_code,roe,profit_dedt,profit_to_gr,assets_turn,saleexp_to_gr,update_flag')
        
        # 合并 + 去重 + 补0
        fina = pd.concat([fina, updated_fina]).sort_values(by=['ts_code', 'update_flag'], ascending=[True, False])
        fina = fina.drop_duplicates(subset=['ts_code']).drop('update_flag', axis=1)

        # 合并利润表数据及财务指标数据
        df = pd.merge(income, fina, on='ts_code', how='left').fillna(0)

        # 重置索引
        df = df.reset_index(drop=True)
        row_count, _ = df.shape

        # 预抓取 StockBasic 进行绑定
        basic_dict = {basic.scode: basic for basic in StockBasic.objects.only('scode', 'sname', 'industry')}

        # 计算单季度营业收入、扣非后净利润
        revenue_offset = collections.defaultdict(int)
        profit_dedt_offset = collections.defaultdict(int)
        if not quarter.endswith('Q1'):
            last_quarter = quarter.replace('Q2', 'Q1').replace('Q3', 'Q2').replace('Y', 'Q3')
            for c, r, p in StockAchievement.objects.filter(
                quarter=last_quarter
            ).values_list(
                'scode', 'revenue', 'profit_dedt'
            ):
                revenue_offset[c] = r
                profit_dedt_offset[c] = p

        for idx, row in df.iterrows():
            callback and callback(idx, row_count)

            if row.ts_code not in basic_dict:
                continue
            
            try:
                basic = basic_dict[row.ts_code]
                StockAchievement.objects.update_or_create(
                    stock_basic=basic,
                    quarter=quarter,
                    defaults={
                        'scode': basic.scode,
                        'sname': basic.sname,
                        'industry': basic.industry,
                        'report_date': datetime.datetime.strptime(row.ann_date, '%Y%m%d').date(),
                        'roe': row.roe,
                        'revenue': row.revenue,
                        'total_cogs': row.total_cogs,
                        'profit_dedt': row.profit_dedt,
                        'assets_turn': row.assets_turn,
                        'profit_to_gr': row.profit_to_gr,
                        'saleexp_to_gr': row.saleexp_to_gr,
                        'single_revenue': row.revenue - revenue_offset[basic.scode],
                        'single_profit_dedt': row.profit_dedt - profit_dedt_offset[basic.scode]
                    }
                )
            except Exception as e:
                Logger.error(msg=f"股票业绩数据保存失败，错误：{e}", module='crawler')

        Logger.info(msg=f"Tushare 业绩筛选 更新已完成，本次更新 {row_count} 只股票记录。", module='crawler')

    
    @classmethod
    def update_all_stock_forecast(cls, period=None, callback=print):
        '''
            方法: 计算并覆盖指定周期的所有股票业绩预告
                参数: 
                    period: 周期
                    callback: 进度更新回调函数
        '''
        Logger.info(msg="Tushare 业绩预告 更新已启动。", module='crawler')

        if isinstance(period, datetime.date):
            period = str(period)
        else:
            try:
                if period in get_quarter_periods():
                    period = period.replace('Q1', '0331').replace('Q2', '0630').replace('Q3', '0930').replace('Y', '1231')
                else:
                    datetime.datetime.strptime(period, '%Y%m%d')
            except Exception as e:
                return Logger.debug(msg=f"Illegal Period {period}, {e}", module='crawler')
        
        last_period = f'{int(period[:4]) - 1}{period[4:]}'
        last_year = f'{last_period[:4]}1231'

        # 周期不合法时退出
        if not any(period.endswith(tail) for tail in ('0331', '0630', '0930', '1231')):
            return Logger.warning(msg=f"Illegal Period {period}.", module='crawler')

        quarter = period.replace('0331', 'Q1').replace('0630', 'Q2').replace('0930', 'Q3').replace('1231', 'Y')
        quarter_index = ('0331', '0630', '0930', '1231').index(period[4:])

        df = pd.concat(
            [
                TS_PRO.forecast_vip(
                    period=period, 
                    type=t,
                    fields='ts_code,ann_date,end_date,type,p_change_min,'
                           'p_change_max,net_profit_min,net_profit_max'
                )
                for t in (["预增", "续盈", "扭亏", "略增"])
            ]
        )

        # 去除脏数据
        df = df.dropna()

        # 去重
        df = df.sort_values(by=['ann_date'], ascending=[False])
        df = df.drop_duplicates(subset=['ts_code'])

        # 上年同期净利润
        last_period_income = TS_PRO.income_vip(period=last_period, fields='ts_code,n_income,update_flag', limit=2**16)

        # 去重 + 补0
        last_period_income = last_period_income.sort_values(by=['ts_code', 'update_flag'], ascending=[True, False]) 
        last_period_income = last_period_income.drop_duplicates(subset=['ts_code']).drop('update_flag', axis=1)

        if last_year == last_period:
            last_year_income = last_period_income
        else:
            # 上年度净利润
            last_year_income = TS_PRO.income_vip(period=last_year, fields='ts_code,n_income,update_flag', limit=2**16)
            
            # 去重 + 补0
            last_year_income = last_year_income.sort_values(by=['ts_code', 'update_flag'], ascending=[True, False]) 
            last_year_income = last_year_income.drop_duplicates(subset=['ts_code']).drop('update_flag', axis=1)

        # 合并上年同期净利润
        df = pd.merge(df, last_period_income, how='left', on='ts_code').rename(columns={'n_income': 'last_period_income'})

        # 合并上年度净利润
        df = pd.merge(df, last_year_income, how='left', on='ts_code').rename(columns={'n_income': 'last_year_income'})

        # 按公布日期批量拉取当日指标信息
        basic = pd.concat(
            [
                TS_PRO.daily_basic(
                    trade_date=ann_date, 
                    ts_code=','.join(df[df.ann_date==ann_date].ts_code.to_list()), 
                    fields='ts_code,pe_ttm,close,total_share,total_mv'
                )
                for ann_date in set(df.ann_date.to_list())
            ]
        )

        # 合并数据
        df = pd.merge(df, basic, how='left', on='ts_code').rename(columns={
            'pe_ttm': 'intraday_pe_ttm', 'close': 'intraday_close',
            'total_share': 'intraday_total_share', 'total_mv': 'intraday_total_mv',
        })

        # 补充年度预测利润
        df['annual_profit_min'] = df.net_profit_min / (quarter_index + 1) * 4
        df['annual_profit_max'] = df.net_profit_max / (quarter_index + 1) * 4
        
        # 补充年度预测PE
        df['annual_pe'] = df.intraday_total_mv / df.annual_profit_min

        # 补充PEG
        df['peg'] = df.intraday_pe_ttm / df.p_change_max

        df = df.reset_index(drop=True).fillna(0).replace([np.inf, -np.inf], -1)
        row_count, _ = df.shape

        # 预抓取 StockBasic 进行绑定
        basic_dict = {basic.scode: basic for basic in StockBasic.objects.only('scode', 'sname', 'industry')}

        for idx, row in df.iterrows():
            callback and callback(idx, row_count)

            try:
                if row.ts_code not in basic_dict:
                    Logger.warning(msg=f"业绩预告未保存，未查询到 `{row.ts_code}` 基础信息。", module='crawler')
                    continue

                basic = basic_dict[row.ts_code]

                row = row.to_dict()
                row.pop('end_date')
                row['scode'] = row.pop('ts_code')
                row['sname'] = basic.sname
                row['industry'] = basic.industry
                row['report_date'] = datetime.datetime.strptime(row.pop('ann_date'), '%Y%m%d').date()

                StockForecast.objects.update_or_create(
                    stock_basic=basic,
                    quarter=quarter,
                    defaults={
                        ** row
                    }
                )

            except Exception as e:
                Logger.error(msg=f"股票业绩数据保存失败，错误：{e}", module='crawler')

        Logger.info(msg=f"Tushare 业绩预告 更新已完成，本次更新 {row_count} 只股票记录。", module='crawler')


    @classmethod
    def update_all_stock_fund_portfolio(cls, period=None, callback=print):
        '''
            方法: 计算并覆盖指定周期的所有公募基金持股
                参数: 
                    period: 周期
                    callback: 进度更新回调函数
        '''
        Logger.info(msg="Tushare 公募基金持股 更新已启动。", module='crawler')

        if isinstance(period, datetime.date):
            period = str(period)
        else:
            try:
                if period in get_quarter_periods():
                    period = period.replace('Q1', '0331').replace('Q2', '0630').replace('Q3', '0930').replace('Y', '1231')
                else:
                    datetime.datetime.strptime(period, '%Y%m%d')
            except Exception as e:
                return Logger.debug(msg=f"Illegal Period {period}, {e}", module='crawler')
        
        # 周期不合法时退出
        if not any(period.endswith(tail) for tail in ('0331', '0630', '0930', '1231')):
            return Logger.warning(msg=f"Illegal Period {period}.", module='crawler')

        quarter = period.replace('0331', 'Q1').replace('0630', 'Q2').replace('0930', 'Q3').replace('1231', 'Y')

        # 根据不同报告选择不同报告日期（因TS不能通过截止日期筛选，此处将截止日期转为报告日期范围）
        if quarter.endswith('Y'):
            start_date, end_date = f'{int(quarter[:4]) + 1}0101', f'{int(quarter[:4]) + 1}0401'
        elif quarter.endswith('Q1'):
            start_date, end_date = f'{quarter[:4]}0401', f'{quarter[:4]}0501'
        elif quarter.endswith('Q2'):
            start_date, end_date = f'{quarter[:4]}0801', f'{quarter[:4]}0901'
        elif quarter.endswith('Q3'):
            start_date, end_date = f'{quarter[:4]}1001', f'{quarter[:4]}1101'

        # 因数据量过大，此处需按批次拉取数据
        df = tushare_batch_fetch(TS_PRO.fund_portfolio, start_date=start_date, end_date=end_date)
        # 过滤截止日期合法项，并派出港股
        df = df[df.end_date==period]
        df = df[~df.symbol.str.endswith('.HK')]

        # 拉取基金基础信息及份额
        fund_basic = tushare_batch_fetch(TS_PRO.fund_basic, fields='ts_code,name,fund_type')
        fund_share = tushare_batch_fetch(TS_PRO.fund_share, trade_date=period, fields='ts_code,fd_share')

        fund = pd.merge(fund_basic, fund_share, on='ts_code', how='left').fillna(0)
        # 合并 去除空数据
        df = pd.merge(df, fund, on='ts_code', how='left').dropna()

        # 拉取股票股本信息
        daily_basic = tushare_batch_fetch(TS_PRO.daily_basic, trade_date=period, fields='ts_code,total_share,float_share')
        daily_basic = daily_basic.rename(columns={'ts_code': 'symbol'})

        # 合并 去除空数据
        df = pd.merge(df, daily_basic, on='symbol', how='left').dropna()

        # 预抓取 StockBasic 进行绑定
        basic_dict = {basic.scode: basic for basic in StockBasic.objects.only('scode', 'sname', 'industry')}

        # 删除数据库中当季度的历史数据
        FundPortfolio.objects.filter(quarter=quarter).delete()

        # 字段重命名
        df = df.rename(columns={
            'ts_code': 'fcode',
            'name': 'fname',
            'symbol': 'scode',
        }).drop(
            ['end_date', 'ann_date'], 
            axis=1
        )

        # 分批次保存数据
        row_count, _ = df.shape
        batch = []
        for idx, row in df.iterrows():
            if row.scode not in basic_dict:
                Logger.warning(msg=f"公募基金持股未保存，未查询到 `{row.scode}` 基础信息。", module='crawler')
                continue
                
            basic = basic_dict[row.scode]

            try:
                row = row.to_dict()
                row['quarter'] = quarter
                row['stock_basic'] = basic
                row['sname'] = basic.sname
                row['industry'] = basic.industry

                batch.append(FundPortfolio(**row))
                if len(batch) > 127: 
                    FundPortfolio.objects.bulk_create(batch)
                    batch.clear()
                
                    callback and callback(idx, row_count)
            
            except Exception as e:
                Logger.error(msg=f"公募基金持股数据保存失败，错误：{e}", module='crawler')

        if batch:
            FundPortfolio.objects.bulk_create(batch)
            batch.clear()

            callback and callback(idx, row_count) 
        
        Logger.info(msg=f"Tushare 公募基金持股 更新已完成，本次更新 {row_count} 只股票记录。", module='crawler')


    @classmethod
    def update_all_stock_consensus_forecasts(cls, callback=print):
        '''
            方法: 计算并覆盖当年一致预期数据
                参数: 
                    callback: 进度更新回调函数
        '''
        Logger.info(msg="Tushare 一致预期 更新已启动。", module='crawler')

        start_date = datetime.date.today() - datetime.timedelta(days=365 * 3)
        start_date = start_date.strftime('%Y%m%d')

        this_year = datetime.date.today().year - 1
        FY0 = f'{this_year}Q4'
        FY1 = f'{this_year + 1}Q4'
        FY2 = f'{this_year + 2}Q4'

        df = tushare_batch_fetch(TS_PRO.report_rc, start_date=start_date, fields='ts_code,report_date,org_name,quarter,np')

        # 仅选取 FY1/FY2 数据
        df = df[(df.quarter == FY1) | (df.quarter == FY2)]

        # 去重 丢弃旧数据
        df = df.sort_values('report_date', ascending=False).drop_duplicates(subset=['ts_code', 'org_name', 'quarter'])
        # 去除空数据
        df = df.dropna()
        
        # 当年净利润
        this_year_income = TS_PRO.income_vip(period=f'{this_year}1231', fields='ts_code,n_income,update_flag', limit=2**16)
        this_year_income = this_year_income.sort_values(by=['ts_code', 'update_flag'], ascending=[True, False]) 
        this_year_income = this_year_income.drop_duplicates(subset=['ts_code']).drop('update_flag', axis=1)

        # 转为字典存储
        income_dict = this_year_income.set_index('ts_code').to_dict(orient='dict')['n_income']

        # 预抓取 StockBasic 进行绑定
        basic_dict = {basic.scode: basic for basic in StockBasic.objects.only('scode', 'sname', 'industry')}

        # 清除历史数据
        ConsensusForecasts.objects.all().delete()

        forecast_objects = []
        ts_codes = set(df.ts_code.to_list())

        for idx, ts_code in enumerate(ts_codes):
            if ts_code not in basic_dict:
                continue

            callback(idx, len(ts_codes))

            try:
                page = df[df.ts_code == ts_code]
                forecast_list = []

                for _, row in page.iterrows():
                    forecast_list.append(row.drop('ts_code').to_dict())

                basic = basic_dict[ts_code]

                np1, np2 = page[page.quarter==FY1].np, page[page.quarter==FY2].np
                
                forecast_objects.append(
                    ConsensusForecasts(
                        scode=ts_code,
                        sname=basic.sname,
                        industry=basic.industry,
                        stock_basic=basic,
                        quarter=FY0.replace('Q4', 'Y'),
                        net_profit_fy0=income_dict.get(ts_code, 0),
                        net_profit_fy1=0 if np1.empty else np1.mean(),
                        net_profit_fy2=0 if np2.empty else np2.mean(),
                        forecast_list=forecast_list
                    )
                )
            except Exception as e:
                Logger.error(msg=f"一致预期数据保存失败，错误：{e}", module='crawler')

        ConsensusForecasts.objects.bulk_create(
            forecast_objects,
            batch_size=32
        )

        Logger.info(msg=f"Tushare 一致预期 更新已完成，本次更新 {len(forecast_objects)} 只股票记录。", module='crawler')


    @classmethod
    def update_all_stock_conceptual(cls, period=None, callback=print):
        '''
            方法: 更新股票所属概念分类季度统计
                参数:
                    callback: 进度更新回调函数
        '''
        Logger.info(msg="Tushare 所属概念分类 更新已启动。", module='crawler')

        quarter = period
        if quarter is None:
            today = datetime.date.today()
            if today.month < 4:
                quarter = f'{today.year - 1}Y'
            elif today.month < 5:
                quarter = f'{today.year}Q1'
            elif today.month < 9:
                quarter = f'{today.year}Q2'
            elif today.month < 11:
                quarter = f'{today.year}Q3'
            else:
                quarter = f'{today.year}Y'
        
        y, q = quarter.replace('Y', 'Q4').split('Q')
        y, q = int(y), int(q)

        qoq_quarter = f'{(y - 1) if q == 1 else y}Q{(q + 3) % 4}'.replace('Q0', 'Y')
        yoy_quarter = f'{y-1}Q{q}'.replace('Q4', 'Y')

        concepts = StockConcept.objects.values_list('ccode', flat=True)
        concepts_count = concepts.count()
        callback(0, concepts_count)

        # 拉取概念清单
        concept_stocks = StockConcept.objects.relation_dataframe()

        '''
        # 预抓取 StockBasic 进行绑定
        basic_dict = {basic.scode: basic for basic in StockBasic.objects.only('scode', 'sname', 'industry')}
        '''
        # 预抓取 StockConcept 进行绑定
        concept_dict = {concept.ccode: concept for concept in StockConcept.objects.only('id', 'ccode', 'cname')}

        # 预抓取同比、环比的营收、扣非利润数据
        ach_queryset = StockAchievement.objects.only('revenue', 'single_revenue', 'profit_dedt', 'single_profit_dedt', 'scode')
        ach_dict = {ach.scode: ach for ach in ach_queryset.filter(quarter=quarter)}
        yoy_ach_dict = {ach.scode: ach for ach in ach_queryset.filter(quarter=yoy_quarter)}
        '''
        qoq_ach_dict = {ach.scode: ach for ach in ach_queryset.filter(quarter=qoq_quarter)}
        
        # 预抓取一致预测数据
        forecast_dict = {
            cf.scode: cf
            for cf in ConsensusForecasts.objects.only(
                'scode', 'net_profit_fy0', 'net_profit_fy1', 'net_profit_fy2'
            )
        }
        '''

        for idx, concept in enumerate(concepts):
            callback(idx, concepts_count)

            try:
                # 拉取概念股票
                stocks = concept_stocks[concept_stocks.ccode==concept].scode.to_list()

                # 当季营收、扣非利润
                revenues = sorted([ach_dict[scode].revenue for scode in stocks if scode in ach_dict])
                profit_dedts = sorted([ach_dict[scode].profit_dedt for scode in stocks if scode in ach_dict])

                # 同比营收、扣非利润
                yoy_revenue = sum([yoy_ach_dict[scode].revenue for scode in stocks if scode in yoy_ach_dict])
                yoy_profit_dedt = sum([yoy_ach_dict[scode].profit_dedt for scode in stocks if scode in yoy_ach_dict])

                '''
                # 当前单季度营收、扣非利润
                single_revenues = sum([ach_dict[scode].single_revenue for scode in df.code.to_list() if scode in ach_dict])
                single_profit_dedts = sum([ach_dict[scode].single_profit_dedt for scode in df.code.to_list() if scode in ach_dict])

                # 环比单季度营收、扣非利润
                qoq_revenues = sum([qoq_ach_dict[scode].single_revenue for scode in df.code.to_list() if scode in qoq_ach_dict])
                qoq_profit_dedts = sum([qoq_ach_dict[scode].single_profit_dedt for scode in df.code.to_list() if scode in qoq_ach_dict])

                # 当年一致预期净利润
                fy0 = sum([forecast_dict[scode].net_profit_fy0 for scode in df.code.to_list() if scode in forecast_dict])
                fy1 = sum([forecast_dict[scode].net_profit_fy1 for scode in df.code.to_list() if scode in forecast_dict])
                fy2 = sum([forecast_dict[scode].net_profit_fy2 for scode in df.code.to_list() if scode in forecast_dict])
                '''
                
                concept_quarter, _ = ConceptQuarter.objects.update_or_create(
                    quarter=quarter,
                    stock_concept=concept_dict[concept],
                    defaults={
                        'revenue': sum(revenues),
                        'revenue_top3': sum(revenues[-3: ]) / (sum(revenues) or float('inf')),
                        'revenue_top5': sum(revenues[-5: ]) / (sum(revenues) or float('inf')),
                        'revenue_yoy': sum(revenues) / (yoy_revenue or sum(revenues) + 1) - 1,
                        # 'revenue_qoq': single_revenues / (qoq_revenues or single_revenues + 1) - 1,
                        'profit_dedt': sum(profit_dedts),
                        # 'profit_dedt_top3': sum(profit_dedts[-3: ]) / (sum(profit_dedts) or float('inf')),
                        # 'profit_dedt_top5': sum(profit_dedts[-5: ]) / (sum(profit_dedts) or float('inf')),
                        'profit_dedt_yoy': sum(profit_dedts) / (yoy_profit_dedt or sum(profit_dedts) + 1) - 1,
                        # 'profit_dedt_qoq': single_profit_dedts / (qoq_profit_dedts or single_profit_dedts + 1) - 1,
                        # 'net_profit_fy0': fy0,
                        # 'net_profit_fy1': fy1,
                        # 'net_profit_fy2': fy2,
                    }
                )
            except Exception as e:
                Logger.debug(msg=f"Conceptual {concept} update failed, {e}", module='crawler')
        
        Logger.info(msg=f"Tushare 所属概念分类季度统计 更新已完成，本次更新 {len(concepts)} 只概念记录。", module='crawler')


    @classmethod
    def update_daily_stock_conceptual(cls, callback=print):
        '''
            方法: 更新股票所属概念分类日级统计
                参数:
                    callback: 进度更新回调函数
        '''
        # 为保证获取的数据至少有60个交易日 起始日期设置为当日的前120日
        today = datetime.date.today()
        preview = 120

        # 拉取个股资金流向
        df = pd.concat([
            tushare_batch_fetch(
                TS_PRO.moneyflow, 
                trade_date=(today - datetime.timedelta(days=offset)).strftime('%Y%m%d'), 
                fields='ts_code,trade_date,net_mf_amount'
            )
            for offset in range(1, preview + 1)
        ])
        # 当前交易日
        trade_date = df.trade_date.max()

        # 预抓取 StockBasic 进行绑定
        basic_dict = {basic.scode: basic for basic in StockBasic.objects.only('id', 'scode', 'sname', 'industry')}

        # 拉取涨跌幅
        daily_change = TS_PRO.daily(
            trade_date=trade_date,
            fields='ts_code,pct_chg'
        )
        # 拉取换手率,流通股本数据
        daily_basic = TS_PRO.daily_basic(
            trade_date=trade_date, 
            fields='ts_code,turnover_rate,float_share'
        )

        # 拉取概念
        concepts = StockConcept.objects.relation_dataframe()
        
        total_count = df.ts_code.drop_duplicates().count()
        total_count += StockConcept.objects.count()
        conceptstock_objects = []
        for sidx, scode in enumerate(df.ts_code.drop_duplicates().to_list()):
            try:
                if daily_basic[daily_basic.ts_code==scode].empty:
                    continue
                if daily_change[daily_change.ts_code==scode].empty:
                    continue
                if scode not in basic_dict:
                    continue 
                
                # 获取该股票的基础信息、涨跌幅、换手率、流通股本数据、资金流向
                stock = basic_dict[scode]
                change = daily_change[daily_change.ts_code==scode].iloc[0]
                basic = daily_basic[daily_basic.ts_code==scode].iloc[0]
                flow = df[df.ts_code==scode]

                conceptstock_objects.append(
                    ConceptStockDaily(
                        scode=scode,
                        sname=stock.sname,
                        industry=stock.industry,
                        concepts='/'.join(concepts[concepts.sid==stock.id].cname),
                        trade_date=date_format(trade_date, str),
                        change_rate=change.pct_chg,
                        turnover_rate=basic.turnover_rate,
                        float_share=basic.float_share,
                        net_mf_d5=flow.net_mf_amount[:5].sum(),
                        net_mf_d20=flow.net_mf_amount[:20].sum(),
                        net_mf_d60=flow.net_mf_amount[:60].sum(),
                        pos_mf_d5_cnt=(flow.net_mf_amount[:5] > 0).sum(),
                        pos_mf_d20_cnt=(flow.net_mf_amount[:20] > 0).sum(),
                        pos_mf_d60_cnt=(flow.net_mf_amount[:60] > 0).sum(),
                    )
                )

                callback(sidx, total_count)
            except Exception as e:
                Logger.debug(msg=f"Daily conceptual Stock {scode} update failed, {e}", module='crawler')
        
        # 写入数据
        ConceptStockDaily.objects.filter(trade_date=date_format(trade_date, str)).delete()
        ConceptStockDaily.objects.bulk_create(conceptstock_objects, batch_size=64)

        # 拉取涨跌幅
        daily_change = TS_PRO.ths_daily(
            trade_date=trade_date,
            fields='ts_code,pct_change'
        )

        # 成交量 = 流通股本 * 换手率
        daily_basic['volume'] = daily_basic.float_share * daily_basic.turnover_rate

        concept_objects = []
        for cidx, concept in enumerate(StockConcept.objects.all()):
            try:
                if daily_change[daily_change.ts_code==concept.ccode].empty:
                    continue
                
                # 获取概念的涨跌幅
                change = daily_change[daily_change.ts_code==concept.ccode].iloc[0]

                # 拉取概念股列表
                stocks = concepts[concepts.ccode==concept.ccode].scode.to_list()
                # 合并成交量,流通股本数据
                total_basic = daily_basic[daily_basic.ts_code.isin(stocks)].sum()

                # 合并资金流向
                flow = df[df.ts_code.isin(stocks)].groupby('trade_date', as_index=False).sum()
                flow = flow.sort_values(by=['trade_date'], ascending=[False])

                if total_basic.float_share > 0:
                    turnover_rate = total_basic.volume / total_basic.float_share / 100
                else:
                    turnover_rate = 0

                concept_objects.append(
                    ConceptDaily(
                        ccode=concept.ccode,
                        cname=concept.cname,
                        stock_basic_count=len(stocks),
                        trade_date=date_format(trade_date, str),
                        change_rate=change['pct_change'],
                        turnover_rate=turnover_rate,
                        net_mf_d5=flow.net_mf_amount[:5].sum(),
                        net_mf_d20=flow.net_mf_amount[:20].sum(),
                        net_mf_d60=flow.net_mf_amount[:60].sum(),
                        pos_mf_d5_cnt=(flow.net_mf_amount[:5] > 0).sum(),
                        pos_mf_d20_cnt=(flow.net_mf_amount[:20] > 0).sum(),
                        pos_mf_d60_cnt=(flow.net_mf_amount[:60] > 0).sum(),
                    )
                )

                callback(sidx + cidx, total_count)
            except Exception as e:
                Logger.debug(msg=f"Daily conceptual {concept.ccode} update failed, {e}", module='crawler')
        
        # 写入数据
        ConceptDaily.objects.filter(trade_date=date_format(trade_date, str)).delete()
        ConceptDaily.objects.bulk_create(concept_objects, batch_size=64)

        Logger.info(msg=f"Tushare 所属概念分类日级统计 更新已完成，本次更新 {len(conceptstock_objects)} 只股票记录，{len(concept_objects)} 只概念记录。", module='crawler')


    @classmethod
    def update_all_stock_fresh_price(cls, callback=print):
        '''
            方法: 更新股票新高
                参数:
                    callback: 进度更新回调函数
        '''
        Logger.info(msg="Tushare 股票新高 更新已启动。", module='crawler')

        stock_count = len(cls.STOCK_CODES)
        # 清空历史
        StockFreshPrice.objects.all().delete()

        # 预抓取 StockBasic 进行绑定
        basic_dict = {basic.id: basic for basic in StockBasic.objects.only('id')}
        objects_list = []

        for idx, info in enumerate(StockBasic.objects.values(
            'scode', 'sname', 'industry', 'id', 'tushare_source'
        )):
            # 拉取源数据
            source = info.pop('tushare_source')
            if not source: continue

            last_date = max(source)

            # 写入 收盘价、持续天数
            info['close'] = source[last_date]['close']
            info['duration_days'] = 1
            # 写入 历史最高
            if len(source) > 1:
                info['extreme'] = info['close'] > max([source[d]['close'] for d in source if d != last_date])
            # 绑定 股票基础信息
            info['stock_basic'] = basic_dict[info.pop('id')]

            # 绑定 股票所属概念
            info['concepts'] = '/'.join(info['stock_basic'].concepts.values_list('cname', flat=True))
            
            # 倒排搜索
            for d in sorted(source, reverse=True):
                # 写入 涨跌幅
                if 'change' not in info and d != last_date:
                    info['change'] = (info['close'] - source[d]['close']) / source[d]['close']
                if d != last_date and source[d]['close'] >= info['close']: 
                    break

                # 持续天数加一
                info['duration_days'] = date_diff(d, last_date) + 1

            objects_list.append(StockFreshPrice(**info))

            callback and callback(idx, stock_count)

        StockFreshPrice.objects.bulk_create(objects_list, batch_size=32)
    
        Logger.info(msg=f"Tushare 股票新高 更新已完成，本次更新 {len(objects_list)} 只概念记录。", module='crawler')


class WindAPI:
    waking_up = False
    keep_running = False
    update_queue = []

    @classproperty
    def STOCK_CODES(cls):
        return TushareAPI.STOCK_CODES

    @classmethod
    def connect(cls, *args, **kwargs):
        '''
            方法： 连接至Wind服务器
                参数：
                    无
        '''
        if cls.waking_up:
            return
        cls.waking_up = True
        w.start(waitTime=120, )
        if w.isconnected():
            Logger.debug(msg="Wind API connected successfully.", module='crawler')
        else:
            Logger.debug(msg="Wind API connected failed.", module='crawler')
        cls.waking_up = False

    @classmethod
    def update_all_stock_basic(cls,  start_date = None, end_date = None, callback = print, period = 'daily'):
        '''
            方法:  更新列表中所有股票数据
                全局变量 (引用自 config 文件):
                    THREADS: 线程数量
                参数:
                    start_date: 开始日期
                    end_date: 结束日期
                    callback: 进度更新回调函数
                    period: 周期
        '''
        assert (start_date and end_date) or (not start_date and not end_date), "Illegal start_date or end_date."
        stock_count = len(cls.STOCK_CODES)
        cls.keep_running = True
        cls.update_queue = cls.STOCK_CODES[:]
        w.start()
        for _ in range(THREADS):
            threading.Thread(target=cls.update_stock_basic, args=(start_date, end_date, period)).start()
        if start_date and end_date:
            Logger.info(msg=f"Wind 源更新已启动，本次更新起止日期：{start_date}~{end_date}。", module='crawler')
        else:
            Logger.info(msg=f"Wind 源更新已启动，本次更新至：{datetime.date.today()}。", module='crawler')
        while cls.keep_running and cls.update_queue:
            if callback is not None: callback(stock_count - len(cls.update_queue), stock_count)
            time.sleep(1)
        cls.keep_running = False
        cls.update_queue.clear()
        Logger.info(msg=f"Wind 源更新完成，本次更新 {stock_count - len(cls.update_queue)} 只股票记录。", module='crawler')

    @classmethod
    def update_all_stock_wbasic(cls,  start_date = None, end_date = None, callback = print,):
        '''
            方法:  更新列表中所有周级股票数据
                全局变量 (引用自 config 文件):
                    THREADS: 线程数量
                参数:
                    start_date: 开始日期
                    end_date: 结束日期
                    callback: 进度更新回调函数
                    period: 周期
        '''
        cls.update_all_stock_basic(start_date, end_date, callback, period = 'weekly')

    @classmethod
    def update_stock_basic(cls, start_date = None, end_date = None, period = 'daily'):
        '''
            子进程方法： 更新列表中所有股票数据(由 update_all_stock_basic 调用)
            参数：
                    start_date: 开始日期
                    end_date: 结束日期
                    period: 周期
        '''
        while cls.keep_running and cls.update_queue:
            code = cls.update_queue.pop()
            for _ in range(RETRY_COUNT):
                try:
                    # 拉取该股票的基础信息记录
                    stock_basic = StockBasic.get_by_code(code)
                    if stock_basic is None: break

                    if start_date and end_date:
                        # 指定日期拉取
                        last_update = datetime.datetime.strptime(str(start_date), "%Y-%m-%d").date()
                        # break
                    else:
                        # 不指定日期时, 更新至最新
                        last_update = datetime.datetime.strptime(
                                max(
                                    (stock_basic.wind_index if period == 'daily' else stock_basic.wind_weekly_index) + [DEFAULT_START_DATE]
                                    ), 
                                "%Y-%m-%d"
                            ).date()
                        end_date = datetime.date.today()
                        # 已更新到最新时跳过
                        if last_update >= end_date: break

                    wind_arg = "PriceAdj=F" + ("" if period == 'daily' else ";Period=W")
                    # 拉取 Wind 前复权的 收盘价 (为计算MA20需设定日期的提前量)
                    status, stock_data = w.wsd(code, "close", str(last_update - datetime.timedelta(days = 40)), str(end_date), wind_arg, usedf = True)
                    # 检查接口返回数据状态
                    if status == WIND_QUOTA_EXCEEDED:
                        cls.keep_running = False
                        Logger.warning(msg=f"Wind 资源超出配额，Wind 源更新已终止。", module='crawler')
                        break
                    
                    stock_data = stock_data.rename(columns = {'CLOSE':'close'})
                    # 去除停牌日 (因 Wind 接口提供数据未过滤停牌日，计算前需先去除停牌日数据)
                    suspend_d = [datetime.datetime.strptime(d, "%Y-%m-%d").date() 
                        for d in stock_basic.suspension 
                        if datetime.datetime.strptime(d, "%Y-%m-%d").date() in stock_data.index
                    ]
                    stock_data = stock_data.drop(labels = suspend_d)

                    # 计算移动平均线
                    for d in ['5', '10', '20']:
                        stock_data['ma' + d] = stock_data['close'].rolling(int(d)).mean()
                    stock_data = stock_data.dropna()

                    # 格式化日期为 2020-12-01 形式
                    stock_data.index = stock_data.index.map(lambda x:str(x))
                    # 去除提前量
                    stock_data = stock_data[stock_data.index > str(last_update)]
                    # 转为 Dict 进行存储
                    stock_data = stock_data.round(4).to_dict(orient = 'index')

                    # 过滤今日收盘前数据
                    if start_date and end_date:
                        if stock_basic.last_update < last_update:
                            stock_basic.last_update = last_update
                    elif datetime.datetime.now().hour < 15:
                        if str(datetime.date.today()) in stock_data:
                            stock_data.pop(str(datetime.date.today()))
                        stock_basic.last_update = datetime.date.today() - datetime.timedelta(days = 1)
                    else:
                        stock_basic.last_update = datetime.date.today()

                    # 筛选后数据为空时跳出
                    if not stock_data: 
                        stock_basic.save()
                        break
                    if period == 'daily':
                        # 增加索引至记录，同时进行排序
                        stock_basic.wind_index = sorted(set(list(stock_data.keys()) + stock_basic.wind_index))
                        # 更新数据
                        stock_basic.wind_source.update(stock_data)
                    else:
                        stock_basic.wind_weekly_index = sorted(set(list(stock_data.keys()) + stock_basic.wind_weekly_index))
                        stock_basic.wind_weekly_source.update(stock_data)
                    
                    stock_basic.save()
                    break
                except Exception as e:
                    Logger.debug(msg=f"Stock {code} update failed, Source: Wind, {e}", module='crawler')
                    stock_data.to_csv(f'BASIC{code}{str(time.time())[:10]}.csv')
                    time.sleep(RETRY_TIME_DELAY)


    @classmethod
    def update_all_stock_rsi(cls,  start_date = None, end_date = None, callback = print, period = 'daily'):
        '''
            方法:  更新列表中所有股票RSI数据
                全局变量 (引用自 config 文件):
                    THREADS: 线程数量
                参数:
                    start_date: 开始日期
                    end_date: 结束日期
                    callback: 进度更新回调函数
                    period: 周期
        '''
        assert (start_date and end_date) or (not start_date and not end_date), "Illegal start_date or end_date."
        stock_count = len(cls.STOCK_CODES)
        cls.keep_running = True
        cls.update_queue = cls.STOCK_CODES[:]
        w.start()
        for _ in range(THREADS):
            threading.Thread(target=cls.update_stock_rsi, args=(start_date, end_date, period)).start()
        if start_date and end_date:
            Logger.info(msg=f"RSI 更新已启动，本次更新起止日期：{start_date}~{end_date}。", module='crawler')
        else:
            Logger.info(msg=f"RSI 源更新已启动，本次更新至：{datetime.date.today()}。", module='crawler')
        while cls.keep_running and cls.update_queue:
            if callback is not None: callback(stock_count - len(cls.update_queue), stock_count)
            time.sleep(1)
        cls.keep_running = False
        cls.update_queue.clear()
        Logger.info(msg=f"RSI 源更新完成，本次更新 {stock_count - len(cls.update_queue)} 只股票记录。", module='crawler')

    @classmethod
    def update_all_stock_wrsi(cls,  start_date = None, end_date = None, callback = print, ):
        '''
            方法:  更新列表中所有股票RSI数据
                全局变量 (引用自 config 文件):
                    THREADS: 线程数量
                参数:
                    start_date: 开始日期
                    end_date: 结束日期
                    callback: 进度更新回调函数
        '''
        cls.update_all_stock_rsi(start_date, end_date, callback, period = 'weekly')

    @classmethod
    def update_stock_rsi(cls, start_date = None, end_date = None, period = 'daily'):
        '''
            子进程方法： 更新列表中所有股票RSI数据
                参数：
                    start_date: 开始日期
                    end_date: 结束日期
                    period: 周期
        '''
        while cls.keep_running and cls.update_queue:
            code = cls.update_queue.pop()
            for _ in range(RETRY_COUNT):
                try:
                    # 拉取该股票的RSI记录
                    stock_rsi = StockRSI.get_by_code(code)
                    if stock_rsi is None: break
                    
                    # 拉取 Wind 前复权的 RSI 数据
                    wind_arg = 'RSI_N=6;PriceAdj=F' + ('' if period ==  'daily' else ';Period=W') 
                    status, stock_data = w.wsd(code, "rsi", str(start_date), str(end_date), wind_arg, usedf = True)
                    # 检查接口返回数据状态
                    if status == WIND_QUOTA_EXCEEDED:
                        cls.keep_running = False
                        Logger.warning(msg="Wind 资源超出配额，RSI 更新已终止。", module='crawler')
                        break

                    # 格式化日期为 2020-12-01 形式
                    stock_data.index = stock_data.index.map(lambda x:str(x))
                    # 去除空数据
                    stock_data = stock_data.dropna()

                    # 跨平台兼容 （Linux 下返回 dataframe 字段为小写）
                    stock_data.rename(columns = {'rsi': 'RSI'}, inplace = True)
                    if 'RSI' not in stock_data.columns:
                        break
                    # 转为 Dict 进行存储
                    stock_data = stock_data.round(4).to_dict(orient = 'dict')['RSI']

                    if period == 'daily':
                        # 增加索引至记录，同时进行排序
                        stock_rsi.rsi_index = sorted(set(list(stock_data.keys()) + stock_rsi.rsi_index))
                        # 更新记录
                        stock_rsi.rsi.update(stock_data)
                    else:
                        stock_rsi.wrsi_index = sorted(set(list(stock_data.keys()) + stock_rsi.wrsi_index))
                        stock_rsi.wrsi.update(stock_data)

                    stock_rsi.save()
                    break
                except Exception as e:
                    Logger.debug(msg=f"Stock {code} update rsi failed, Source: Wind, {e}", module='crawler')
                    stock_data.to_csv(f'RSI{code}{str(time.time())[:10]}.csv')
                    time.sleep(RETRY_TIME_DELAY)
